package com.disruptor.disruptor;

import com.disruptor.bean.DataBean;
import com.disruptor.util.HelpMe;
import com.jfinal.plugin.activerecord.Db;
import com.jfinal.plugin.activerecord.Record;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.lmax.disruptor.util.DaemonThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by mengdexuan on 2017/9/18 10:46.
 *
 * 生产者消费者处理
 */
public class ProConMain {

	private static Logger log = LoggerFactory.getLogger(ProConMain.class);

	// The factory for the event
	private static DataBeanFactory factory = new DataBeanFactory();

	// Construct the Disruptor
	private static Disruptor<DataBean> disruptor = null;
	private static RingBuffer<DataBean> ringBuffer = null;

	//单线程池
	private static ExecutorService executorService = Executors.newSingleThreadExecutor();

	public static void start(int bufferSize,int consumerThreadNum){

		disruptor = new Disruptor(factory,bufferSize, DaemonThreadFactory.INSTANCE, ProducerType.SINGLE, new BlockingWaitStrategy());

		//添加消费者线程
		disruptor.handleEventsWithWorkerPool(consumerThread(consumerThreadNum));

		//设置异常处理器
		disruptor.setDefaultExceptionHandler(new DataBeanExceptionHandler());

		//启动 disruptor
		disruptor.start();

		ringBuffer = disruptor.getRingBuffer();

		producer();
	}

	private static DataBeanHandler[] consumerThread(int num){
		if (num<=0)throw new RuntimeException("disruptor 消费者线程必须大于0，且是 2 的整次幂！");
		DataBeanHandler result[] = new DataBeanHandler[num];
		for (int i=0;i<num;i++){
			result[i] = new DataBeanHandler();
		}
		return result;
	}

	public static void stop(){
		stopProducer();
		try {
			Thread.sleep(1000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		executorService.shutdown();//关闭生产线程池
		disruptor.shutdown();//会消费掉环形队列中的所有数据，然后再执行关闭
		log.info("终止 Disruptor 调度……");
	}

	//启动生产者
	public static void startProducer(){
		producerFlag = true;
	}

	//停止生产者
	public static void stopProducer(){
		producerFlag = false;
	}

	//生产标识
	private static boolean producerFlag = false;


	//生产者状态
	public static boolean isStartProducer(){
		return producerFlag;
	}


	/**
	 * 并发生产数据的问题：同一条数据被多个进程（或线程）读取到，重复生产，用如下方式解决：
	 *
	 * 1.当启动多个 jvm 进程，去处理同一数据库某张表的数据，生产数据时的查询可以用 for update 防止同一条数据被处理多次（利用数据库自身的锁机制）
	 *
	 * 2.利用分布式锁，获取到锁的进程才可以生产数据（redis,zookeeper等都可以实现分布式锁，也有现成的开源实现可用），
	 * 此种方式避免了数据库锁的使用，使用普通的sql查询语句，可以提高数据库性能，也在一定程度上避免了数据库死锁的产生，优于第 1 种方式
	 *
	 * * */
	private static void producer(){
		Runnable task = new Runnable() {
			@Override
			public void run() {
				DataBeanProducer producer = new DataBeanProducer(ringBuffer);

				while (true){

					if (producerFlag){
						List<Record> list = Db.find("select * from test where status = 1 limit 2 for update");
						if (HelpMe.isNotNull(list)){
							for (Record record:list){
								Db.update("update test set status = 2 where id = "+record.getInt("id"));
								//如果 bufferSize 已满，则生产线程会被阻塞
								producer.publishData(record);
							}
						}
						try {
							Thread.sleep(1000);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}else {
						try {
							Thread.sleep(100);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}

				}
			}
		};

		executorService.execute(task);

		startProducer();
	}












}












































